/*
 * note:
 * 	-change in constructor from eclipse - Nick
 */
package rpc;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.PortUnreachableException;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import session.Session;
import session.SessionID;
import appserver.Constants;
import appserver.ServletUtils;
public class RPCClient {
	
	AtomicInteger callIdCount = new AtomicInteger(1);
	Set<String> mbrSet;
	String servletInfo;
	int rpcServerPort;
	
	public RPCClient(Set<String> mbrSet, String servletInfo, int rpcServerPort) {
		this.mbrSet = mbrSet;
		this.servletInfo = servletInfo;
		this.rpcServerPort = rpcServerPort;
	}
	
	public Session sessionRead(String sessionId, int sessVersion, String[] rpcServersInfo) {
		DatagramSocket socket = null;
		try {
			boolean openedSocket = false;
			int port = 0;
			while (!openedSocket) {
				port = ServletUtils.getRandomNumber(32000, 64000);
				socket = null;
				try {
					socket = new DatagramSocket(port);
					socket.setSoTimeout(Constants.SOCKET_TIME_OUT);
					openedSocket = true;
				} catch (SocketException e3) {
				  //try again
				}
			}
			ByteArrayOutputStream bStream = new ByteArrayOutputStream();
			ObjectOutput oo = null;
			try {
				oo = new ObjectOutputStream(bStream);
			} catch (IOException e) {
				System.out.println("Could not open object stream");
				return null;
			} 
			int callIdNumber = 10000 * port + callIdCount.getAndIncrement();
			RPCPacket packet = new RPCPacket(Integer.toString(callIdNumber));
			packet.setRpcPort(rpcServerPort);
			packet.setOpCode(OpCodes.SESSION_READ);
			packet.setSessVersion(sessVersion);
			packet.setSessId(sessionId);
			try {
				oo.writeObject(packet);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			try {
				oo.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			byte[] serializedMessage = bStream.toByteArray();
			System.out.println("Client - sitting on port: " + port);
			for (String server : rpcServersInfo) {
				if (server.equals(Constants.IP_NULL) || server.equals(servletInfo)) continue;
				String[] serverInfo = server.split(Constants.PORT_NUMBER_SEPARATOR);
		        InetAddress IPAddress = null;
				try {
					IPAddress = InetAddress.getByName(serverInfo[0]);
				} catch (UnknownHostException e3) {
					continue;
				}
		        int serverPort = Integer.parseInt(serverInfo[1]);
		        byte[] sendData = new byte[1024];
		        byte[] receiveData = new byte[1024];
		        sendData = serializedMessage;
		        DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, IPAddress, serverPort);
		        boolean packetSent = false; int sendTries = 0;
		        while (!packetSent && sendTries++ < Constants.MAX_TRIES) {
			        try {
			        	System.out.println("Client- sending packet to:" + IPAddress + Constants.PORT_NUMBER_SEPARATOR + serverPort);
						socket.send(sendPacket);
						packetSent = true;
					} catch (IOException e2) {
						System.out.println("Client - Failed sending packet to: " + server);
					}
		        }
		        if (!packetSent) {
		        	mbrSet.remove(server);
		        	continue;
		        }
		        DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
		        boolean receivedPacket = false; int receiveTries = 0;
		        while (!receivedPacket && receiveTries++ < Constants.MAX_TRIES) {
			        try {
			        	System.out.println("Client - expecting response from: " + server);
						socket.receive(receivePacket);
						receivedPacket = true;
					} 
			        catch (PortUnreachableException e) {
			        	System.out.println("Client - port unreachable exception");
			        }
			        
			        catch (SocketTimeoutException e2)  {
						System.out.println("Client - response from server timed out: " + server);
					}
			        catch (IOException e1) {
						System.out.println("Client - Failed receiving response from: " + server);
					}
		        }
		        if (!receivedPacket) {
		        	mbrSet.remove(server);
		        	continue;
		        }
		        ObjectInputStream iStream = null;
				try {
					iStream = new ObjectInputStream(new ByteArrayInputStream(receivePacket.getData()));
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
		        RPCPacket rpcResponse = null;
				try {
					rpcResponse = (RPCPacket) iStream.readObject();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ClassNotFoundException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				if (!rpcResponse.getCallId().equals(Integer.toString(callIdNumber))) {
					continue;
				}
				else if (rpcResponse.isAck()) {
					Session readSession = rpcResponse.getSessionData();
					return readSession;
				}
				else {
					continue;
				}
			}
	        return null;
		}
		catch (Exception e) {
			//do nothing
		}
		finally {
			if (socket != null) socket.close();
		}
		return null;
        }
	
	//    SessionWrite( SID, new_change_count, new_data, discard_time ) 
	public String sessionWrite(SessionID sessId, int newSessVersion, Session sessionData, long discardTime) {
		DatagramSocket socket = null;
		try {
			int port = ServletUtils.getRandomNumber(32000, 64000);
			//boolean[] wasSuccessful = new boolean[rpcServersInfo.length];
			try {
				socket = new DatagramSocket(port);
				socket.setSoTimeout(Constants.SOCKET_TIME_OUT);
			} catch (SocketException e3) {
				// TODO Auto-generated catch block
				e3.printStackTrace();
			}
			ByteArrayOutputStream bStream = new ByteArrayOutputStream();
			@SuppressWarnings("unused")
			ObjectOutput oo = null;
			try {
				oo = new ObjectOutputStream(bStream);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 
			int callIdNumber = 10000 * port + callIdCount.getAndIncrement();
			RPCPacket packet = new RPCPacket(Integer.toString(callIdNumber));
			packet.setRpcPort(rpcServerPort);
			packet.setOpCode(OpCodes.SESSION_WRITE);
			packet.setSessionData(sessionData);
			try {
			oo.writeObject(packet);
			oo.close();
			}
			catch (IOException e) {
				
			}
			byte[] serializedRpcPacket = bStream.toByteArray();
			//while loop
			int mbrSize = mbrSet.size();
			//Iterator<String> mbrIt = mbrSet.iterator();
			//int chosenMbr = ServletUtils.getRandomNumber(0, mbrSize);
			boolean sessionStored = false; 
			while (mbrSize > 1 && !sessionStored) {
				mbrSize = mbrSet.size();
				Iterator<String> mbrIt = mbrSet.iterator();
				int chosenMbr = ServletUtils.getRandomNumber(0, mbrSize);
				for (int i = 0; i <= chosenMbr || !mbrIt.hasNext(); i++) {
					String currentServer = mbrIt.next();
					if (chosenMbr != i) continue;
					if (currentServer.equals(servletInfo)) break;
					String[] serverInfo = currentServer.split(Constants.PORT_NUMBER_SEPARATOR);
					InetAddress serverIp = null;
					try {
					serverIp = InetAddress.getByName(serverInfo[0]);
					}
					catch (UnknownHostException e) {
						System.out.println("Removing unrecognized server IPP form mbrSet:" + currentServer);
						mbrSet.remove(currentServer);
						break;
					}
					//send
			        DatagramPacket sendPacket = new DatagramPacket(serializedRpcPacket, serializedRpcPacket.length, serverIp, Integer.parseInt(serverInfo[1]));
				    boolean packetSent = false; int sendTries = 0;
				    while (!packetSent && sendTries++ < Constants.MAX_TRIES) {
			        	try {
							socket.send(sendPacket);
							packetSent = true;
						} catch (IOException e) {
							System.out.println("Could not send packet to:" + currentServer);
						}
				    }
				    if (!packetSent) {
				    	System.out.println("Failed sending packet to: " + currentServer);
				    	mbrSet.remove(currentServer);
				    	break;
				    }
			        byte[] receiveData = new byte[1024];
			        DatagramPacket receivePacket = new DatagramPacket(receiveData, receiveData.length);
			        boolean packetReceived = false; int receivedTries = 0;
			        while (!packetReceived && receivedTries++ < Constants.MAX_TRIES) {
				        try {
							socket.receive(receivePacket);
							packetReceived = true;
						} catch (IOException e) {
						}
			        }
			        if (!packetReceived) {
			        	System.out.println("Failed to obtain response from: " + currentServer);
			        	mbrSet.remove(currentServer);
			        	break;
			        }
			        //receive ack
			        ObjectInputStream iStream = null;
			        RPCPacket info = null;
			        try {
				        try {
							iStream = new ObjectInputStream(new ByteArrayInputStream(receivePacket.getData()));
						} catch (IOException e) {
							System.out.println("Error creating object input stream");
							break;
						}
				        try {
							info = (RPCPacket) iStream.readObject();
						} catch (IOException e) {
							System.out.println("Error creating object input stream");
							break;
						}
			        }
			        catch (ClassNotFoundException e) {
			        	System.out.println("Invalid request (not an RPCPacket)");
			        	break;
			        }
			        try {
						iStream.close();
					} catch (IOException e) {
						System.out.println("Error closing object stream");
			        	break;
					}
			        if (info.getCallId() == null || !info.getCallId().equals(Integer.toString(callIdNumber))) {        
			        	break;
			        }
			        else {
			        	if (info.isAck()) {
			        		return currentServer;
			        	}
			        }
				}
			}
			return null;
		}
		catch (Exception e) {
			//do nothing
		}
		finally {
			if (socket != null) socket.close();
		}
		return null;
	}
	
	
	public void delete(SessionID sid, String rpcInfo) {
		if (rpcInfo.equals(Constants.IP_NULL)) return;
		DatagramSocket socket = null;
		try {
			int port = ServletUtils.getRandomNumber(32000, 64000);
			try {
				socket = new DatagramSocket(port);
				socket.setSoTimeout(Constants.SOCKET_TIME_OUT);
			} catch (SocketException e3) {
				// TODO Auto-generated catch block
				e3.printStackTrace();
			}
			ByteArrayOutputStream bStream = new ByteArrayOutputStream();
			@SuppressWarnings("unused")
			ObjectOutput oo = null;
			try {
				oo = new ObjectOutputStream(bStream);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} 
			int callIdNumber = 10000 * port + callIdCount.getAndIncrement();
			RPCPacket packet = new RPCPacket(Integer.toString(callIdNumber));
			packet.setRpcPort(rpcServerPort);
			packet.setOpCode(OpCodes.SESSION_DELETE);
			packet.setSessId(sid.toString());
			try {
			oo.writeObject(packet);
			oo.close();
			}
			catch (IOException e) {
			}
			byte[] serializedRpcPacket = bStream.toByteArray();
			String[] rpcInfoArray = rpcInfo.split(Constants.PORT_NUMBER_SEPARATOR);
			if (rpcInfoArray.length != 2) return;
			DatagramPacket sendPacket = new DatagramPacket(serializedRpcPacket, serializedRpcPacket.length, InetAddress.getByName(rpcInfoArray[0]), Integer.parseInt(rpcInfoArray[1]));
			try {
				socket.send(sendPacket);
			}
			catch (IOException e) {
				System.out.println("Client delete(..) - could not send packet to:" + rpcInfo);
			} 
		}
		catch (Exception e) {
			//do nothing
		}
		finally {
			if (socket != null) socket.close();
		}
	}
}
